package com.ccyang;


import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * @author: yangjinpeng
 * @date: 2018-12-23
 * @description: Stream 运行机制
 * 1、所有操作都是链式操作，一个元素只执行一次
 * 2、每一个中间操作返回一个新的流，流里面有一个属性 sourceStage，指向同一个地方，就是Head
 * 3、Head -> nextStage -> nextStage-> ... -> null
 * 4、有状态操作会把无状态操作截断，单独处理。
 * 5、并行环境下，有状态的中间操作，不一定能并行处理。
 * 6、parallel / sequential 这2个操作也是中间操作(也是返回stream)
 *    但是他们不创建流, 他们只修改 Head的并行标志
 *
 */
public class RunStream {

    public static void main(String[] args) {

        Random random = new Random();
        // 随机产生数据
        Stream<Integer> stream = Stream.generate(() -> random.nextInt())
                // 产生500个元素
                .limit(500)
                // 第一个无状态操作
                .peek(s -> print("peek: " + s))
                // 第二个无状态操作
                .filter(
                        s -> {
                            print("filter: " + s);
                            return s > 100000;
                        }
                )
                // 有状态操作
                .sorted((s1, s2) -> {
                    print("排序：" + s1 + ", " + s2);
                    return s1.compareTo(s2);
                })
                // 无状态操作
                .peek(s -> print("peek2: " + s))
                .parallel();

        stream.count();
    }

    // 打印日志，并且sleep 5毫秒
    private static void print(String str){

        // System.out.println(s);
        // 带线程名(测试并行情况)
        System.out.println(Thread.currentThread().getName() + " > " + str);
        try {
            TimeUnit.MILLISECONDS.sleep(5);
        } catch (InterruptedException e) { }
    }

}
